package xin.marcher.wind.migrate.migrate;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.LocalDateTimeUtil;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;
import xin.marcher.framework.common.util.SnowflakeIdWorker;
import xin.marcher.framework.common.util.spring.SpringContextUtil;
import xin.marcher.wind.migrate.component.lock.ScrollLock;
import xin.marcher.wind.migrate.constants.MigrateConstant;
import xin.marcher.wind.migrate.domain.RangeScroll;
import xin.marcher.wind.migrate.domain.ScrollInfo;
import xin.marcher.wind.migrate.domain.entity.EtlDirtyRecordDO;
import xin.marcher.wind.migrate.domain.entity.EtlProgressDO;
import xin.marcher.wind.migrate.domain.entity.EtlStatisticalDO;
import xin.marcher.wind.migrate.enums.EtlProgressStatus;
import xin.marcher.wind.migrate.mapper.MigrateScrollMapper;
import xin.marcher.wind.migrate.service.MigrateConfigService;
import xin.marcher.wind.migrate.service.MigrateService;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.List;
import java.util.Map;

/**
 * 全量滚动数据处理
 */
@Slf4j
@Component
public class ScrollProcessor {

    /**
     * 抽取数据的大小
     */
    private static final Integer PAGE_SIZE = 500;

    /**
     * 滚动锁对象
     */
    private final ScrollLock lock = new ScrollLock();

    @Resource
    private MigrateScrollMapper migrateScrollMapper;

    @Resource
    MigrateConfigService migrateConfigService;

    /**
     * 全量同步数据入口(可指定从某个批次的序号开始再次执行)
     *
     * @param rangeScroll 数据抽取模型
     */
    public void scroll(RangeScroll rangeScroll) {
        boolean lockState = this.lock.lock(rangeScroll.getTableName(), rangeScroll.getDomain());
        if (lockState) {
            try {
                //初始化总数据量统计容器
                if (ObjectUtils.isEmpty(rangeScroll.getPageSize())) {
                    rangeScroll.setPageSize(PAGE_SIZE);
                }
                if (ObjectUtils.isEmpty(rangeScroll.getTicket())) {
                    rangeScroll.setTicket(SnowflakeIdWorker.generateIdStr());
                }
                if (StringUtils.isEmpty(rangeScroll.getStartScrollId())) {
                    MigrateService migrateService = SpringContextUtil.getBean(MigrateService.class);
                    String scrollId = migrateService.queryMinScrollId(rangeScroll);
                    rangeScroll.setStartScrollId(scrollId);
                }
                // 没有符合条件的数据，此次滚动不处理
                if (StringUtils.isEmpty(rangeScroll.getStartScrollId())) {
                    log.info("没有符合条件的数据, 迁移取消");
                    return;
                }

                // 填充唯一滚动字段
                String singleKey = MergeConfig.getSingleKey(rangeScroll.getTableName());
                rangeScroll.setScrollName(singleKey);
                rangeScroll.setTargetTableName(migrateConfigService.getSourceTableName(rangeScroll.getDomain(), rangeScroll.getTableName()));
                // 保存迁移计划
                EtlProgressDO etlProgressDO = addEtlProgress(rangeScroll);

                // 对滚动的数据进行处理
                scrollDate(etlProgressDO, rangeScroll);
            } catch (Exception e) {
                log.error("滚动拉取数据异常{}", JSONObject.toJSONString(rangeScroll), e);
            } finally {
                lock.unlock(rangeScroll.getTableName(), rangeScroll.getDomain());
            }
        }
    }

    /**
     * 循环滚动数据
     *
     * @param etlProgressDO 数据抽取批次
     */
    private void scrollDate(EtlProgressDO etlProgressDO, RangeScroll rangeScroll) {
        EtlDirtyRecordDO etlDirtyRecordDO = null;
        try {
            MigrateService migrateService = SpringContextUtil.getBean(MigrateService.class);
            // 滚动抽取数据 当抽取完最后一批数据后将同步状态为同步完成
            List<Map<String, Object>> queryInfoList = migrateService.queryInfoList(rangeScroll);
            while (CollectionUtils.isNotEmpty(queryInfoList)) {
                // 数据同步
                MergeBinlogWrite mergeBinlogWrite = new MergeBinlogWrite();
                // 拿当前的这批数据，标记最后一条数据的关键分页字段更新
                ScrollInfo scrollInfo = mergeBinlogWrite.load(queryInfoList, rangeScroll);

                // 当批量写入的数据为 0 时，可能已经在时间范围内同步完成
                // 这个时候查询进度数据，如果进度达到 100%，则更新当前的任务为完成状态
                if (checkEtlProgressSuccess(scrollInfo, etlProgressDO)) {
                    //更新当前的同步任务为同步完成
                    updateEtlProgressSuccess(etlProgressDO);
                    return;
                }

                // 初始化本批次的明细数据
                etlDirtyRecordDO = insertEtlDirtyRecord(etlProgressDO, rangeScroll);

                rangeScroll.setStartScrollId(scrollInfo.getMaxScrollId());
                etlProgressDO.setScrollId(scrollInfo.getMaxScrollId());
                etlDirtyRecordDO.setSyncSize(scrollInfo.getScrollSize());
                // 更新这个批次的数据同步完成
                updateEtlProgress(etlProgressDO, etlDirtyRecordDO, EtlProgressStatus.SUCCESS.getValue(), scrollInfo.getScrollSize());
                etlProgressDO.setCurTicketStage(etlProgressDO.getCurTicketStage() + 1);
                // 继续滚动抽取数据
                queryInfoList = migrateService.queryInfoList(rangeScroll);
            }

            updateEtlProgressSuccess(etlProgressDO);
        } catch (Exception e) {
            log.error("循环滚动数据错误", e);
            if (null != etlProgressDO) {
                updateEtlProgressFail(etlProgressDO, etlDirtyRecordDO, 0);
            }
        }
    }

    /**
     * 初始化记录本次迁移数据明细
     *
     * @param rangeScroll 数据抽取模型
     * @return 迁移表记录
     */
    private EtlProgressDO addEtlProgress(RangeScroll rangeScroll) {
        EtlProgressDO etlProgressDO = new EtlProgressDO();

        etlProgressDO.setDomain(rangeScroll.getDomain());
        etlProgressDO.setDomainId(getDomainId(rangeScroll.getDomain()));
        etlProgressDO.setLogicModel(rangeScroll.getTableName());
        if (ObjectUtils.isEmpty(rangeScroll.getCurTicketStage())) {
            etlProgressDO.setCurTicketStage(1);
        } else {
            etlProgressDO.setCurTicketStage(rangeScroll.getCurTicketStage() + 1);
        }
        etlProgressDO.setTicket(rangeScroll.getTicket());
        if (!ObjectUtils.isEmpty(rangeScroll.getRetryTimes())) {
            etlProgressDO.setRetryTimes(rangeScroll.getRetryTimes());
        } else {
            etlProgressDO.setRetryTimes(0);
        }

        // 断点续传的时候，同步 finishRecord 数据过来，然后将翻页的页数改为默认值，
        // 防止当同步了几百万数据时，一次取百万数据导致异常
        if (PAGE_SIZE.equals(rangeScroll.getPageSize())) {
            etlProgressDO.setFinishRecord(0);
        } else {
            etlProgressDO.setFinishRecord(rangeScroll.getPageSize());
            rangeScroll.setPageSize(PAGE_SIZE);
        }
        etlProgressDO.setProgressType(0);
        etlProgressDO.setStatus(EtlProgressStatus.INIT.getValue());
        etlProgressDO.setScrollId(rangeScroll.getStartScrollId());
        etlProgressDO.setScrollTime(rangeScroll.getStartTime());
        etlProgressDO.setScrollEndTime(rangeScroll.getEndTime());
        etlProgressDO.setCreateTime(new DateTime());
        etlProgressDO.setUpdateTime(new DateTime());
        // 重试直接返回
        if (rangeScroll.getRetryFlag()) {
            etlProgressDO.setProgressType(rangeScroll.getProgressType());
            rangeScroll.setRetryFlag(false);
            return etlProgressDO;
        } else {
            migrateScrollMapper.insertEtlProgress(etlProgressDO);
        }
        return etlProgressDO;
    }

    /**
     * 新增数据迁移的进度明细信息
     *
     * @param etlProgressDO
     */
    private EtlDirtyRecordDO insertEtlDirtyRecord(EtlProgressDO etlProgressDO, RangeScroll rangeScroll) {
        EtlDirtyRecordDO etlDirtyRecordDO = new EtlDirtyRecordDO();
        etlDirtyRecordDO.setLogicModel(etlProgressDO.getLogicModel());
        etlDirtyRecordDO.setCurTicketStage(etlProgressDO.getCurTicketStage());
        etlDirtyRecordDO.setTicket(etlProgressDO.getTicket());
        etlDirtyRecordDO.setStatus(EtlProgressStatus.INIT.getValue());
        etlDirtyRecordDO.setRetryTimes(0);
        etlDirtyRecordDO.setSyncSize(rangeScroll.getPageSize());
        etlDirtyRecordDO.setRecordValue(etlProgressDO.getScrollId());
        etlDirtyRecordDO.setRecordKey(MergeConfig.getSingleKey(etlProgressDO.getLogicModel()));
        etlDirtyRecordDO.setCreateTime(new DateTime());
        etlDirtyRecordDO.setUpdateTime(new DateTime());
        migrateScrollMapper.insertEtlDirtyRecord(etlDirtyRecordDO);

        return etlDirtyRecordDO;
    }

    /**
     * 更新 本次迁移数据的状态为失败
     *
     * @param etlProgressDO 迁移表记录
     */
    private void updateEtlProgressFail(EtlProgressDO etlProgressDO, EtlDirtyRecordDO etlDirtyRecordDO, Integer pageSize) {
        etlProgressDO.setStatus(EtlProgressStatus.FAIL.getValue());
        updateEtlProgress(etlProgressDO, etlDirtyRecordDO, EtlProgressStatus.FAIL.getValue(), pageSize);
    }

    /**
     * 更新 本次迁移数据的状态
     *
     * @param etlProgressDO 迁移表记录
     */
    private void updateEtlProgress(EtlProgressDO etlProgressDO, EtlDirtyRecordDO etlDirtyRecordDO, Integer status, Integer pageSize) {
        etlProgressDO.setUpdateTime(new DateTime());
        etlProgressDO.setFinishRecord(etlProgressDO.getFinishRecord() + pageSize);
        etlProgressDO.setScrollId(etlProgressDO.getScrollId());
        migrateScrollMapper.updateEtlProgress(etlProgressDO);
        etlDirtyRecordDO.setCurTicketStage(etlProgressDO.getCurTicketStage());
        // 更新本次明细的状态
        updateEtlDirtyRecord(etlDirtyRecordDO, status);
    }

    /**
     * 更新 本次迁移数据的状态为成功
     *
     * @param etlProgressDO 迁移表记录
     */
    public void updateEtlProgressSuccess(EtlProgressDO etlProgressDO) {
        etlProgressDO.setUpdateTime(new DateTime());
        etlProgressDO.setStatus(EtlProgressStatus.SUCCESS.getValue());
        migrateScrollMapper.updateEtlProgress(etlProgressDO);
    }

    /**
     * 当批量写入的数据为0时，可能已经在时间范围内同步完成
     * 这个时候查询进度数据，如果进度达到100%，则更新当前的任务为完成状态
     * 缓存当前批次的数据总量
     *
     * @param scrollInfo    滚动信息
     * @param etlProgressDO 迁移表记录
     */
    public boolean checkEtlProgressSuccess(ScrollInfo scrollInfo, EtlProgressDO etlProgressDO) {
        if (scrollInfo.getScrollSize() == 0) {
            BigDecimal statisticalCount;
            if (MigrateConstant.statisticalCountMap.containsKey(etlProgressDO.getTicket())) {
                // 获取已同步的数据（通过 CountCacheTask 分天统计计算的数据）
                statisticalCount = MigrateConstant.statisticalCountMap.get(etlProgressDO.getTicket());
            } else {
                Integer startTime = Integer.valueOf(LocalDateTimeUtil.format(etlProgressDO.getScrollTime(), DatePattern.PURE_DATE_PATTERN));
                Integer endTime = Integer.valueOf(LocalDateTimeUtil.format(etlProgressDO.getScrollEndTime(), DatePattern.PURE_DATE_PATTERN));

                EtlStatisticalDO etlStatisticalDO = new EtlStatisticalDO();
                etlStatisticalDO.setDomain(etlProgressDO.getDomain());
                etlStatisticalDO.setLogicModel(etlProgressDO.getLogicModel());
                etlStatisticalDO.setStartTime(startTime);
                etlStatisticalDO.setEndTime(endTime);
                etlStatisticalDO.setDomain(etlProgressDO.getDomain());
                // 获取已同步的数据（通过 CountCacheTask 分天统计计算的数据）
                statisticalCount = migrateScrollMapper.getStatisticalCount(etlStatisticalDO);
                MigrateConstant.statisticalCountMap.put(etlProgressDO.getTicket(), statisticalCount);
            }
            if (new BigDecimal(etlProgressDO.getFinishRecord().toString()).compareTo(statisticalCount) >= 0) {
                //更新当前的同步任务为同步完成
                updateEtlProgressSuccess(etlProgressDO);
                return true;
            } else {
                return false;
            }
        }
        return false;
    }

    /**
     * 更新本次迁移数据的明细状态
     *
     * @param etlDirtyRecordDO
     * @param status
     */
    private void updateEtlDirtyRecord(EtlDirtyRecordDO etlDirtyRecordDO, Integer status) {
        etlDirtyRecordDO.setUpdateTime(new DateTime());
        etlDirtyRecordDO.setStatus(status);
        etlDirtyRecordDO.setRetryTimes(etlDirtyRecordDO.getRetryTimes() + 1);
        migrateScrollMapper.updateEtlDirtyRecord(etlDirtyRecordDO);
    }

    private Long getDomainId(String domain) {
        return migrateConfigService.getDomainId(domain);
    }
}
